/**
 * 与 MQTT WebWorker 搭配，提供一套基于 rxjs 的 MQTT 连接操作类
 */
import { getLogger } from '@gnlab/utils'
import { MemStream, mmap, mdistinct } from '@gnlab/utils/rx'
import memoize from 'lodash/memoize'
import { Subject, Observable } from 'rxjs'
import { multicast, refCount } from 'rxjs/operators'
import type {
  ConnStatus,
  MQTTMessages,
  SubscribeResult,
  UnsubscribeResult,
} from '@/lib/mqtt/worker'
import ConnWorker from '@/lib/mqtt/worker?worker'
import { WorkerContext } from '@/lib/web-worker'

const logger = getLogger('lib/mqtt')

export interface Message {
  topic: string
  payload: string
}

export default class Connection {
  constructor(
    public address: string,
    public qos = 1 as 0 | 2 | 1,
  ) {
    this.connect()
    this.initTopicSubscription()
  }

  /**
   * 初始化连接
   */
  protected connWorker = new WorkerContext<MQTTMessages['host'], MQTTMessages['worker']>(ConnWorker)
  protected connect() {
    this.connWorker.post('connect', { address: this.address, qos: this.qos })
  }

  /**
   * 记录连接状态
   */
  status = new MemStream<ConnStatus>('connecting', subscriber => {
    this.connWorker.on('status-change', ({ current, prev }) => {
      if (current === 'connecting' || current === 'reconnecting') logger.info(current)
      if (current === 'connected')
        logger.info(prev === 'reconnecting' ? 'reconnected' : 'connected')
      if (current === 'connectFailed')
        logger.info(
          prev === 'connecting' || prev === 'reconnecting' ? 'reconnect failed' : 'lost connection',
        )
      subscriber.next(current)
    })
  })

  /**
   * 当前是否已连接
   */
  connected = this.status.pipe(
    mmap(status => status === 'connected'),
    mdistinct(),
  )

  /**
   * 订阅消息
   */

  /*
  返回一个 observable，监听指定 topic 的消息
  - 此 observable 被订阅时，同步进行 MQTT topic 订阅
  - 此 observable 取消订阅时，同步取消 MQTT topic 订阅
  */
  topic = memoize((id: string) => {
    const observable = new Observable<Message>(subscriber => {
      function handleMessage(msg: Message) {
        if (msg.topic.startsWith(id)) subscriber.next(msg)
      }

      this.connWorker.on('message', handleMessage)
      // MQTT 订阅或取消订阅都必须等 MQTT 处于连接成功的状态时才能执行
      // （严格来说，还要考虑订阅/取消订阅行为本身是异步的，以及有操作失败的可能性。但目前还没有因为这些情况遇到过问题，因此忽略它们）
      this.status.subscribe(status => {
        if (status === 'connected') this.connWorker.post('subscribe', id)
      })

      return () => {
        this.connWorker.off('message', handleMessage)
        this.status.subscribe(status => {
          if (status === 'connected') this.connWorker.post('unsubscribe', id)
        })
      }
    })

    // MQTT.js 中无论执行多少次订阅行为，都只会实际订阅一次；然后只要取消一次，就会真正取消。
    // 但我们应用中，一个 topic 的消息要被多个业务模块监听，因此通过 multicast() 和 refCount() 结合，
    // 实现在第一次监听时，触发 MQTT 订阅；在最后一次取消监听时，才取消 MQTT 订阅。
    return observable.pipe(multicast(new Subject()), refCount())
  })

  protected initTopicSubscription() {
    function handleSubscribeResult(result: SubscribeResult) {
      if (result.success) logger.info(`subscribe success: ${result.topic}`, result.info)
      else logger.info(`subscribe failed: ${result.topic}`, result.error)
    }
    function handleUnsubscribeResult(result: UnsubscribeResult) {
      if (result.success) logger.info(`unsubscribe success: ${result.topic}`, result.info)
      else logger.info(`unsubscribe failed: ${result.topic}`, result.error)
    }
    this.connWorker.on('subscribe-result', handleSubscribeResult)
    this.connWorker.on('unsubscribe-result', handleUnsubscribeResult)
  }

  /**
   * 推送消息
   */
  publish(topicId: string, payload: string) {
    logger.debug('send message', topicId, payload)
    this.connWorker.post('publish', { topic: topicId, payload })
  }
}
